package org.ns.tomdog.server.nioprotocol;

import org.ns.tomdog.server.*;
import org.ns.tomdog.server.support.ServerResponseImpl;
import org.ns.tomdog.utils.HttpParse;

import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

public class NioProtocolHandler implements ProtocolHandler {

    private ServerSocketChannel serverSocketChannel;
    private Selector selector;
    private Thread selectorThread;
    private Thread clientReadThread;
    private ExecutorService businessThreads;
    private LinkedBlockingDeque<SelectionKey> socketChannels = new LinkedBlockingDeque<>();
    private ConnectorConfig config;
    private ProtocolAdapter protocolAdapter;

    @Override
    public void setProtocolAdapter(ProtocolAdapter protocolAdapter) {
        this.protocolAdapter = protocolAdapter;
    }

    @Override
    public void setConnectionConfig(ConnectorConfig config) {
        this.config = config;
    }

    @Override
    public void init() {
        try {
            int port = config.getPort();
            int maxConnection = config.getMaxConnection();
            int maxThread = config.getMaxThread();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(port), maxConnection);
            serverSocketChannel.configureBlocking(false);
            selector = Selector.open();
            selectorThread = new Thread( this::runSelector, "Connector-Selector");
            clientReadThread = new Thread(this::runConsume, "Connector-Consume");
            // selectorThread.setDaemon(true);
            businessThreads = Executors.newFixedThreadPool(maxThread);
            System.out.println("绑定端口:" + port);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void start() {
        try {
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            selectorThread.start();
            clientReadThread.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private void runSelector() {
        while (serverSocketChannel.isOpen()) {
            try {
                selector.select(3000);
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();

                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    if (!selectionKey.isValid()) {
                        continue;
                    }

                    if (selectionKey.isAcceptable()) {
                        ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel();
                        SocketChannel accept = server.accept();
                        accept.configureBlocking(false);
                        accept.register(selector, SelectionKey.OP_READ);
                    }
                    if (selectionKey.isReadable()) {
                        socketChannels.add(selectionKey);
                        selectionKey.cancel();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        }
    }
    private void runConsume() {
        while (serverSocketChannel.isOpen()) {
            try {
                SelectionKey selectionKey = socketChannels.poll(3, TimeUnit.SECONDS);

                if (selectionKey != null) {
                    final SocketChannel client = (SocketChannel) selectionKey.channel();
                    client.configureBlocking(true);
                    businessThreads.execute(()->{
                        try {

                            handler(client);
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            try {
                                System.out.println("close--------");
                                client.close();
                                selectionKey.cancel();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    });
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void handler(SocketChannel client) throws Exception{
        InputStream inputStream = Channels.newInputStream(client);
        OutputStream outputStream = Channels.newOutputStream(client);
        ServerRequest serverRequest = HttpParse.parse(inputStream);
        ServerResponse serverResponse = new ServerResponseImpl(outputStream);
        try {
            protocolAdapter.service(serverRequest, serverResponse);
        } catch (Exception e) {
            serverResponse.setStatus(500);
            serverResponse.setVersion(serverRequest.getVersion());
            serverResponse.setError("SERVER_ERROR");
            throw e;
        } finally {
            serverResponse.flush();
        }
    }
}
